package com.powernode.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.powernode.dao.ProdEsDao;
import com.powernode.domain.Prod;
import com.powernode.model.ProdEs;
import com.powernode.model.ProdEsChange;
import com.powernode.service.ImportService;
import com.powernode.service.ProdService;
import com.powernode.utils.ProductThreadPool;
import com.rabbitmq.client.Channel;
import io.swagger.models.auth.In;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 内容:
 * 作者: cy
 */
@Service
@Slf4j
@RefreshScope
public class ImportServiceImpl implements ImportService, CommandLineRunner {

    @Autowired
    private ProdEsDao prodEsDao;

    @Autowired
    private ProdService prodService;

    @Value("${importes.size}")
    private Integer size;

    private Date t1;

    /**
     * 全量导入
     * 触发？ 项目启动
     * 将mysql的数据 同步到es去
     * 分批分页 多线程的方式导入
     * 自定义分页  总条数 当前页  每页条数
     */
    @Override
    public void importAll() {
        log.info("全量导入开始");
        // 查询总条数
        int totalCount = prodService.selectCount(null, null);
        if (totalCount == 0) {
            log.info("没有数据需要导入");
            return;
        }
        // 计算页数
        int totalPage = totalCount % this.size == 0 ? totalCount / this.size : ((totalCount / this.size) + 1);
        // 创建一个线程计数器
        CountDownLatch countDownLatch = new CountDownLatch(totalPage);

        for (int i = 1; i <= totalPage; i++) {
            int current = i;
            // 线程提交任务 异步去执行导入
            ProductThreadPool.poolExecutor.execute(() -> {
                importProdToEs(current, this.size, null, null);
                // 计数器 --
                countDownLatch.countDown();
            });
        }

        try {
            countDownLatch.await(5, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 这个时间会比 导入完成的时间要小
        t1 = new Date();
    }

    /**
     * 查询数据库 导入es的方法
     *
     * @param current
     * @param size
     * @param t1
     * @param t2
     */
    private void importProdToEs(int current, Integer size, Date t1, Date t2) {
        // 分页是发两条sql语句 一个是查总条数 一个是查当前页数据
        Page<Prod> page = new Page<>(current, size, false);
        // 查询数据 组装符合条件的商品对象
        List<ProdEs> prodEsList = prodService.selectProdByEs(page, t1, t2);
        //   List<Prod> ----> List<ProdEs>
        // 导入
        prodEsDao.saveAll(prodEsList);
    }

    /**
     * 增量 因为增量的数据 也可能很多 一样需要使用分批分页多线程的方式
     * 查询 updateTime
     * 增删改的数据 我们叫增量数据
     * 定时任务  10分钟 执行一次   做性能和实时性的综合
     * 为了测试方便 2min
     * initialDelay 第一次什么时候执行
     * fixedRate 第一次以后  每次执行的间隔
     */
    @Override
    @Scheduled(initialDelay = 120 * 1000, fixedRate = 120 * 1000)
    public void importUpdate() {
        log.info("增量导入开始");
        Date t2 = new Date();
        // 查询总条数
        int totalCount = prodService.selectCount(t1, t2);
        if (totalCount == 0) {
            t1 = t2;
            log.info("没有数据需要导入");
            return;
        }
        // 计算页数
        int totalPage = totalCount % this.size == 0 ? totalCount / this.size : ((totalCount / this.size) + 1);
        // CountDownLatch  --     CyclicBarrier ++ 最后一个线程 还需要执行一个任务   Semaphore 原子计数器  ++ --
        CountDownLatch countDownLatch = new CountDownLatch(totalPage);

        for (int i = 1; i <= totalPage; i++) {
            int current = i;
            // 线程提交任务 异步去执行导入
            ProductThreadPool.poolExecutor.execute(() -> {
                importProdToEs(current, this.size, t1, t2);
                countDownLatch.countDown();
            });
        }

        try {
            countDownLatch.await(5, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        t1 = t2;
    }

    /**
     * 快速
     * 比如说 有些场景  需要立刻去同步es
     * 可以去提供一个手动触发方案
     * 或者提供这样的功能 其他模块可以使用这样的功能
     * 我们定制化一个  修改库存的方案
     * 使用mq来做一个快速方案
     */
    @Override
    public void importQuick() {
        // this method is empty
    }


    /**
     * 定义一个方法
     * 来做快速导入的实现
     * 这个导入 就是 修改库存的方案
     * order-service----扣减es库存---->product-service
     * prodId
     * count
     * 先定义好这样的数据
     * mq来做业务的前提 就是参数需要定义好
     * <p>
     * Map<ProdId,Count>
     * ProdEsChange{
     * prodId  18        18
     * count (+-)  -1    1
     * }
     * prodEsChange
     * List<ProdEsChange>  适配多个商品的修改
     *
     * @param message
     * @param channel
     */
    //    @RabbitListener(queues = "")
    public void handlerQuickMsg(Message message, Channel channel) {
        // 拿到数据
        String msgStr = new String(message.getBody());
        List<ProdEsChange> prodEsChanges = JSON.parseArray(msgStr, ProdEsChange.class);
        // 修改es的库存
        List<Long> prodIds = prodEsChanges.stream()
                .map(ProdEsChange::getProdId)
                .collect(Collectors.toList());
        // 查es
        Iterable<ProdEs> prodEsIterable = prodEsDao.findAllById(prodIds);
        // 循环es集合
        prodEsIterable.forEach(prodEs -> {
            ProdEsChange prodEsChange1 = prodEsChanges.stream()
                    .filter(prodEsChange -> prodEsChange.getProdId().equals(prodEs.getProdId()))
                    .collect(Collectors.toList())
                    .get(0);
            // 特别注意 这里使用 + - 来做操作的判断 减库存  负1   回滚就传 正1
            long finalCount = prodEs.getTotalStocks() + prodEsChange1.getCount();
            if (finalCount < 0) {
                log.error("库存不足,商品id为{}", prodEs.getProdId());
            }
            prodEs.setTotalStocks(finalCount);
            prodEs.setSoldNum(prodEs.getSoldNum() - prodEsChange1.getCount());
        });
        // 批量修改
        prodEsDao.saveAll(prodEsIterable);
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 整个spring容器加载完以后执行这个方法
     *
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        importAll();
    }


    //    @PostConstruct
//    public void initImport() {
//        importAll();
//    }
}
